package com.atguigu.flink.cep;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2022/12/20
 *
 *  模式序列。
 *          多个模式。
 *          指定多个模式，如何进行串联(连接)
 *
 *
 *     CEP 对原理基于 NFA 设计
 */
public class Demo4_PatternSequence
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, r) -> e.getTs());


        SingleOutputStreamOperator<WaterSensor> ds =
            env.readTextFile("data/cep.txt")
               .map(new WaterSensorMapFunction())
               .assignTimestampsAndWatermarks(watermarkStrategy);


        //第一个模式
        Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("s1")
            .where(new SimpleCondition<WaterSensor>()
            {
                //定义匹配规则
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s1".equals(value.getId());
                }
            })
            //第二个模式对名字
            //  数据的id必须是s1,接下来紧挨着对数据的id必须是s2。 输出两个模式匹配的数据。  严格连续
            //.next("s2")
            //  数据的id必须是s1,接下来紧挨着对数据的id必须不是s2。 输出前一个模式匹配的数据。  严格不连续
            //.notNext("s2")
            //  数据的id必须是s1,接下来 后续有一条数据的id是s2。 输出两个模式匹配的数据   松散连续
            //.followedBy("s2")
            //  数据的id必须是s1,接下来 会和后续的任意一条数据的id是s2对配对。 输出两个模式匹配的数据   松散连续
            //.followedByAny("s2")
            //  数据的id必须是s1,接下来 后续的任意一条数据的id不是s2。 输出两个模式匹配的数据   松散不连续
            //  NotFollowedBy is not supported as a last part of a Pattern（不能作为条件对最后一部分，可以作为中间部分）!  流的数据是源源不断对来，无法保证后续对数据是否一定不是s2
            // 数据必须是s1，后续必须跟有s3，中间不能有s3
            .notFollowedBy("s2")
            .where(new SimpleCondition<WaterSensor>()
            {
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s2".equals(value.getId());
                }
            })
            .followedBy("s3")
            .where(new SimpleCondition<WaterSensor>()
            {
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "s3".equals(value.getId());
                }
            });






        PatternStream<WaterSensor> patternStream = CEP.pattern(ds, pattern);


        patternStream.select(new PatternSelectFunction<WaterSensor, String>()
        {

            @Override
            public String select(Map<String, List<WaterSensor>> map) throws Exception {
                return map.toString();
            }
        })
                     .print().setParallelism(1);


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
